package com.sloan.community.service;

import com.sloan.community.common.Constant;
import com.sloan.community.common.command.CommentCommand;
import com.sloan.community.common.dto.NoticeDTO;
import com.sloan.community.common.vo.CommentVO;
import com.sloan.community.common.vo.DelCommentVO;
import com.sloan.community.exception.CommunityException;
import com.sloan.community.po.*;
import com.sloan.community.repository.*;
import com.sloan.community.util.TextUtils;
import com.sloan.community.util.TimeUtils;
import com.sloan.community.ws.CommunityWebSocketHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.context.annotation.Lazy;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.sloan.community.common.Constant.*;
import static com.sloan.community.common.Constant.CommentStatus.HIDE;
import static com.sloan.community.common.Constant.CommentStatus.SHOW;
import static com.sloan.community.common.Constant.NoticeStatus.UN_READ;
import static com.sloan.community.common.Constant.NoticeType.ARTICLE;
import static com.sloan.community.common.Constant.NoticeType.COMMENT;

@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy})
public class CommentService {

    private final CommentRepository commentRepository;
    private final UserRepository userRepository;
    private final CommentLikeCountRepository commentLikeCountRepository;
    private final ArticleRepository articleRepository;
    private final NotificationRepository notificationRepository;
    private final CommunityWebSocketHandler webSocketHandler;

    public Flux<CommentVO> articleCommentList(Long articleId, User user) {
        Mono<Article> articleMono = articleRepository.findById(articleId)
                .switchIfEmpty(articleNotFoundException(articleId))
                .cache();

        Flux<Comment> dbComments = commentRepository.findByArticleId(articleId)
                .switchIfEmpty(Flux.empty())
                .cache();

        Mono<Map<Long, Comment>> commentMapMono = dbComments.collectMap(Comment::getId, Function.identity());

        Mono<Map<Long, HashSet<Long>>> userLikeMapMono = this.getUserLikeMapMono(articleId);
        Mono<Map<Long, User>> userMapMono = this.getUserMapMono(dbComments);

        Flux<CommentVO> commentResFlux = Mono.zip(dbComments.collectList(),
                        userMapMono, userLikeMapMono, commentMapMono, articleMono)
                .flatMapMany(tuple -> {
                    List<Comment> comments = tuple.getT1();
                    Map<Long, User> userMap = tuple.getT2();
                    Map<Long, HashSet<Long>> userLikeMap = tuple.getT3();
                    Map<Long, Comment> commentMap = tuple.getT4();
                    return Flux.fromIterable(comments.stream()
                            .map(comment -> transfer(user, comment, userMap,
                                    tuple.getT5(), userLikeMap, commentMap))
                            .toList());
                }).cache();

        Flux<CommentVO> rootComments = commentResFlux
                .filter(c -> Objects.equals(PARENT_ROOT_ID, c.getParentCommentId()))
                .cache();

        // flatMap 和 flatMany内部可能会产生多线程进行流式操作，导致乱序
        Flux<CommentVO> commentVoFlux = rootComments
                .flatMap(comment -> {
                    List<CommentVO> childrenList = new ArrayList<>();
                    comment.setChildrenComments(childrenList);
                    return dfsComment(comment, commentResFlux, childrenList);
                })
                .collectList()// 等所有元素都到达后在排序
                .flatMapMany(comments -> {
                    // 等到这里同步排序
                    comments.forEach(comment -> comment.getChildrenComments()
                            .sort(Comparator.comparing(CommentVO::getLikeCount, Comparator.reverseOrder())
                                    .thenComparing(CommentVO::getCreatedTime)));

                    comments.sort(Comparator
                            .comparing(CommentVO::getLikeCount, Comparator.reverseOrder())
                            .thenComparing(CommentVO::getCreatedTime));
                    return Flux.fromIterable(comments);
                });

        return Mono.when(articleMono)
                .thenMany(commentVoFlux);
    }

    private Mono<Map<Long, HashSet<Long>>> getUserLikeMapMono(Long articleId) {
        return commentLikeCountRepository.findByArticleId(articleId)
                .groupBy(CommentLikeCount::getCommentId, CommentLikeCount::getUserId)
                .flatMap(group -> group.collectList()
                        .map(item -> new AbstractMap.SimpleEntry<>(group.key(), new HashSet<>(item))))
                .collectList()
                .map(item -> item.stream()
                        .collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey,
                                AbstractMap.SimpleEntry::getValue, (k1, k2) -> k1)));
    }

    private Mono<Map<Long, User>> getUserMapMono(Flux<Comment> dbComments) {
        Flux<Long> fromUserIds = dbComments.map(Comment::getUserId).distinct();
        Flux<Long> replyUserIds = dbComments.map(Comment::getReplyUserId).distinct();
        Mono<Set<Long>> userIdsMono = Mono.zip(fromUserIds.collectList(), replyUserIds.collectList())
                .map(tuple -> {
                    Set<Long> list = new HashSet<>();
                    list.addAll(tuple.getT1());
                    list.addAll(tuple.getT2());
                    return list;
                });

        return userIdsMono.flatMap(userIds -> userRepository.findByIdIsIn(userIds)
                .collectMap(User::getId, Function.identity()));
    }


    public Mono<CommentVO> dfsComment(CommentVO commentVO, Flux<CommentVO> commentFlux, List<CommentVO> childrenList) {
        Flux<CommentVO> childrenFlux = commentFlux
                .filter(comment -> comment.getParentCommentId().equals(commentVO.getCommentId()))
                .cache();

        Flux<CommentVO> findCommentFlux = childrenFlux
                .flatMap(comment -> dfsComment(comment, commentFlux, childrenList));

        Mono<CommentVO> mono = childrenFlux.collectList()
                .map(childrenComments -> {
                    childrenList.addAll(childrenComments);
                    return commentVO;
                });

        return Mono.zip(mono, findCommentFlux.collectList())
                .map(Tuple2::getT1);
    }


    @Transactional(rollbackFor = Exception.class)
    public Mono<CommentVO> add(Long articleId, User user, CommentCommand commentCommand) {
        Mono<Article> articleMono = articleRepository.findById(articleId)
                .switchIfEmpty(articleNotFoundException(articleId))
                .<Article>handle((article, sink) -> {
                    if (!article.getCommentable()) {
                        sink.error(CommunityException.of("当前文章不可以评论"));
                        return;
                    }
                    sink.next(article);
                })
                .cache();

        Mono<User> userMono = Mono.just(Optional.ofNullable(commentCommand.getReplyUserId())
                        .orElseGet(() -> {
                            commentCommand.setReplyUserId(PARENT_ROOT_ID);
                            return PARENT_ROOT_ID;
                        }))
                .flatMap(replyUserId -> {
                    if (!PARENT_ROOT_ID.equals(replyUserId)) {
                        return userRepository.findById(replyUserId)
                                .switchIfEmpty(Constant.userNotFoundException(commentCommand.getReplyUserId()));
                    }
                    return Mono.just(new User());
                })
                .cache();

        Mono<Comment> commentMono = Mono.just(Optional.ofNullable(commentCommand.getParentCommentId())
                        .orElse(PARENT_ROOT_ID))
                .flatMap(parentCommentId -> {
                    if (!PARENT_ROOT_ID.equals(parentCommentId)) {
                        return commentRepository.findById(parentCommentId)
                                .switchIfEmpty(Mono.error(CommunityException.of(HttpStatus.NOT_FOUND, "回复评论不存在"
                                        + commentCommand.getParentCommentId())));
                    }
                    return Mono.just(new Comment());
                })
                .cache();

        return userMono
                .then(commentMono)
                .then(articleMono)
                .flatMap(article -> {
                    Comment comment = new Comment();
                    comment.setUserId(user.getId());
                    comment.setArticleId(articleId);
                    comment.setParentCommentId(commentCommand.getParentCommentId());
                    Long replyUserId = PARENT_ROOT_ID.equals(commentCommand.getReplyUserId()) ? article.getUserId()
                            : commentCommand.getReplyUserId();
                    comment.setReplyUserId(replyUserId);
                    comment.setContent(commentCommand.getContent());
                    comment.setStatus(SHOW.getCode());
                    comment.setLikeCount(0L);
                    return commentRepository.save(comment)
                            .flatMap(dbComment -> {
                                // 新增通知
                                return insertNotification(article, user, dbComment);
                            });
                })
                .flatMap(comment -> {
                    Mono<User> authorMono = userRepository.findById(comment.getUserId());
                    // 这里就可以用Mono.zip了,因为上面的 commentMono, articleMono, userMono
                    //   通过Mono.then()已经控制了错误流程，如果发生错误，就不会走到这里,走到这里说明没有问题
                    NoticeDTO noticeDTO = new NoticeDTO();
                    noticeDTO.setToUserId(comment.getReplyUserId());
                    noticeDTO.setFormUserId(comment.getUserId());
                    Mono<Void> voidMono = webSocketHandler.pushMsg(noticeDTO);
                    return voidMono.then(Mono.zip(authorMono, commentMono, articleMono, userMono)
                            .map(tuple -> {
                                User author = tuple.getT1();
                                User replyUser = tuple.getT1();
                                Comment replyComment = tuple.getT2();
                                return getAddCommentVO(comment, author, tuple.getT3(),
                                        replyUser, replyComment);
                            }));
                });
    }

    private Mono<Comment> insertNotification(Article article, User user, Comment dbComment) {
        Long formUserId = user.getId();
        boolean isRoot = PARENT_ROOT_ID.equals(dbComment.getParentCommentId());
        Long toUserId = isRoot ? article.getUserId() : dbComment.getReplyUserId();
        if (formUserId.equals(toUserId)) {
            return Mono.just(dbComment);
        }
        Notification notification = new Notification();
        notification.setFromUserId(formUserId);
        notification.setToUserId(toUserId);
        notification.setArticleId(article.getId());
        notification.setCommentId(dbComment.getId());
        notification.setStatus(UN_READ.getFlag());
        notification.setType(isRoot ? ARTICLE.getCode() : COMMENT.getCode());
        return notificationRepository.save(notification)
                .thenReturn(dbComment);
    }

    private CommentVO getAddCommentVO(Comment comment, User author, Article article,
                                      User replyUser, Comment replyComment) {
        CommentVO addCommentVO = new CommentVO();
        BeanUtils.copyProperties(comment, addCommentVO);
        addCommentVO.setCommentId(comment.getId());
        addCommentVO.setSessionUserLike(Boolean.FALSE);
        addCommentVO.setAuthor(comment.getUserId().equals(article.getUserId()));
        addCommentVO.setCommentable(article.getCommentable());
        this.handleReplyComment(replyComment, addCommentVO);

        addCommentVO.setOwner(Boolean.TRUE);
        addCommentVO.setDateTime(TimeUtils.timeDiff(comment.getCreatedTime()));
        addCommentVO.setUserId(author.getId());
        addCommentVO.setUsername(author.getUsername());
        addCommentVO.setAvatar(author.getAvatarUrl());
        addCommentVO.setNickname(author.getNickname());
        addCommentVO.setAddress(author.getAddress());
        addCommentVO.setAuthFrom(author.getAuthFrom());
        addCommentVO.setFormT(Constant.Oauth2.getByCode(author.getAuthFrom()).getMsg());

        if (comment.getUserId().equals(comment.getReplyUserId())) {
            replyUser = author;
        }
        addCommentVO.setReplyUserId(replyUser.getId());
        addCommentVO.setReplyUsername(replyUser.getUsername());
        addCommentVO.setReplyAvatar(replyUser.getAvatarUrl());
        addCommentVO.setReplyNickname(replyUser.getNickname());
        addCommentVO.setReplyAddress(replyUser.getAddress());
        addCommentVO.setReplyAuthFrom(replyUser.getAuthFrom());
        addCommentVO.setReplyFormT(Constant.Oauth2.getByCode(replyUser.getAuthFrom()).getMsg());
        return addCommentVO;
    }


    private void handleReplyComment(Comment replyComment, CommentVO commentVO) {
        if (Objects.nonNull(replyComment) && StringUtils.hasText(replyComment.getContent())) {
            String replyText = replyComment.getContent().replaceAll(IMG_TAG, IMG_TEXT);
            commentVO.setReplyContent(TextUtils.handleImgHtmlToText(replyText, REPLY_COMMENT_LENGTH));
        }
    }

    @Transactional(rollbackFor = Exception.class)
    public Mono<Long> likeCount(Long commentId, User user) {
        Mono<Comment> commentMono = commentRepository.findById(commentId)
                .switchIfEmpty(Constant.commentNotFoundException(commentId))
                .handle((comment, sink) -> {
                    if (HIDE.getCode().equals(comment.getStatus())) {
                        sink.error(CommunityException.of("当前评论已删除,不可以在点赞"));
                        return;
                    }
                    sink.next(comment);
                });

        return commentLikeCountRepository.findByCommentIdIsIn(List.of(commentId))
                .map(CommentLikeCount::getUserId)
                .distinct()
                .filter(likeUserId -> likeUserId.equals(user.getId()))
                .count()
                .flatMap(count -> {
                    if (count == 0) {
                        return commentLike(commentId, user, commentMono);
                    }
                    return commentUnLike(commentId, user, commentMono);
                });
    }

    public Mono<DelCommentVO> delete(Long commentId, User user) {
        return commentRepository.findById(commentId)
                .switchIfEmpty(Constant.commentNotFoundException(commentId))
                .flatMap(comment -> {
                    if (!user.getId().equals(comment.getUserId())) {
                        return Mono.error(CommunityException.of("非法操作"));
                    }
                    comment.setStatus(HIDE.getCode());
                    return commentRepository.save(comment);
                }).then(Mono.fromSupplier(() -> {
                    DelCommentVO commentVO = new DelCommentVO();
                    commentVO.setCommentId(commentId);
                    commentVO.setMsg(COMMENT_DEL_MSG);
                    return commentVO;
                }));
    }

    public Mono<Long> commentLike(Long commentId, User user, Mono<Comment> commentMono) {
        Mono<Void> voidMono = commentRepository.likeCountById(commentId);
        Mono<CommentLikeCount> commentLikeCountMono = commentMono.flatMap(comment -> {
            CommentLikeCount commentLikeCount = new CommentLikeCount();
            commentLikeCount.setCommentId(commentId);
            commentLikeCount.setUserId(user.getId());
            commentLikeCount.setArticleId(comment.getArticleId());
            return commentLikeCountRepository.save(commentLikeCount);
        });
        return voidMono
                .then(commentLikeCountMono)
                .then(commentMono.map(Comment::getLikeCount));
    }

    public Mono<Long> commentUnLike(Long commentId, User user, Mono<Comment> commentMono) {
        Mono<Void> voidMono = commentRepository.unLikeCountById(commentId);
        Mono<Void> mono = commentLikeCountRepository.deleteByCommentIdAndUserId(commentId, user.getId());
        return voidMono.then(mono)
                .then(commentMono.map(Comment::getLikeCount));
    }

    public Flux<CommentVO> commentChat(Long commentId, User user) {
        Mono<Comment> commentMono = commentRepository.findById(commentId).cache();
        Flux<Comment> commentFlux = commentMono
                .switchIfEmpty(Constant.commentNotFoundException(commentId))
                .flatMapMany(comment -> dfsCommentChat(comment, new ArrayList<>()))
                .cache();

        Mono<Article> articleMono = commentMono
                .flatMap(comment -> articleRepository.findById(comment.getArticleId()));

        Mono<Map<Long, User>> userMapMono = getUserMapMono(commentFlux);

        Mono<Map<Long, HashSet<Long>>> userLikeMapMono = commentFlux.map(Comment::getId).collectList()
                .flatMapMany(commentLikeCountRepository::findByCommentIdIsIn)
                .groupBy(CommentLikeCount::getCommentId, CommentLikeCount::getUserId)
                .flatMap(group -> group.collectList()
                        .map(item -> new AbstractMap.SimpleEntry<>(group.key(), new HashSet<Long>(item))))
                .collectMap(Map.Entry::getKey, Map.Entry::getValue);

        return Mono.zip(commentFlux.collectList(), userMapMono, userLikeMapMono, articleMono)
                .flatMapMany(tuple -> {
                    List<CommentVO> list = tuple.getT1().stream()
                            .map(comment -> transfer(user, comment, tuple.getT2(),
                                    tuple.getT4(), tuple.getT3(), null))
                            .toList();
                    return Flux.fromIterable(list);
                });
    }

    public Flux<Comment> dfsCommentChat(Comment comment, List<Comment> comments) {
        comments.add(0, comment);
        if (PARENT_ROOT_ID.equals(comment.getParentCommentId())) {
            return Flux.fromIterable(comments);
        }

        return commentRepository.findById(comment.getParentCommentId())
                .flatMapMany(item -> dfsCommentChat(item, comments));
    }

    private CommentVO transfer(User user, Comment comment, Map<Long, User> userMap, Article article,
                               Map<Long, HashSet<Long>> userLikeMap, Map<Long, Comment> commentMap) {
        Long parentCommentId = comment.getParentCommentId();
        CommentVO commentVO = new CommentVO();
        commentVO.setArticleId(comment.getArticleId());
        commentVO.setCommentId(comment.getId());
        commentVO.setOwner(Objects.nonNull(user) && comment.getUserId().equals(user.getId()));
        commentVO.setParentCommentId(parentCommentId);
        if (comment.getStatus().equals(HIDE.getCode())) {
            commentVO.setContent(COMMENT_DEL_MSG);
        }
        if (comment.getStatus().equals(SHOW.getCode())) {
            commentVO.setContent(comment.getContent());
        }

        if (Objects.nonNull(commentMap) && !PARENT_ROOT_ID.equals(parentCommentId)
                && commentMap.containsKey(parentCommentId)) {
            Comment replyComment = commentMap.get(parentCommentId);
            if (replyComment.getStatus().equals(HIDE.getCode())) {
                commentVO.setReplyContent(COMMENT_DEL_MSG);
                commentVO.setReplyIdStatus(Boolean.FALSE);
            }
            if (replyComment.getStatus().equals(SHOW.getCode())) {
                this.handleReplyComment(replyComment, commentVO);
                commentVO.setReplyIdStatus(Boolean.TRUE);
            }
        }

        commentVO.setStatus(comment.getStatus());
        commentVO.setLikeCount(comment.getLikeCount());
        commentVO.setCreatedTime(comment.getCreatedTime());
        commentVO.setDateTime(TimeUtils.timeDiff(commentVO.getCreatedTime()));
        commentVO.setSessionUserLike(Objects.nonNull(user)
                && userLikeMap.getOrDefault(comment.getId(), new HashSet<>()).contains(user.getId()));
        commentVO.setAuthor(comment.getUserId().equals(article.getUserId()));
        commentVO.setCommentable(article.getCommentable());

        User fromUser = userMap.get(comment.getUserId());
        commentVO.setUserId(comment.getUserId());
        commentVO.setUsername(fromUser.getUsername());
        commentVO.setAvatar(fromUser.getAvatarUrl());
        commentVO.setNickname(fromUser.getNickname());
        commentVO.setAddress(fromUser.getAddress());
        commentVO.setAuthFrom(fromUser.getAuthFrom());
        commentVO.setFormT(Constant.Oauth2.getByCode(fromUser.getAuthFrom()).getMsg());

        User replayUser = userMap.get(comment.getReplyUserId());
        commentVO.setReplyUserId(comment.getReplyUserId());
        commentVO.setReplyUsername(replayUser.getUsername());
        commentVO.setReplyAvatar(replayUser.getAvatarUrl());
        commentVO.setReplyNickname(replayUser.getNickname());
        commentVO.setReplyAddress(replayUser.getAddress());
        commentVO.setReplyAuthFrom(replayUser.getAuthFrom());
        commentVO.setReplyFormT(Constant.Oauth2.getByCode(replayUser.getAuthFrom()).getMsg());
        return commentVO;
    }
}
